﻿// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

#if !NET40
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using CuteAnt.AsyncEx;
using CuteAnt.Runtime;

namespace CuteAnt.IO.Pipelines
{
  internal sealed class PipelineStream : Stream
  {
    private const int c_initialSize = 1024 * 64;
    private readonly static Task<int> _initialCachedTask = Task.FromResult(0);
    private Task<int> _cachedTask = _initialCachedTask;

    private readonly IPipe _pipe;
    private readonly WritableBuffer _output;

    public PipelineStream(IPipe pipe, int initialSize = c_initialSize)
    {
      _pipe = pipe;
      _output = _pipe.Writer.Alloc(initialSize);
    }

    public override bool CanRead => true;

    public override bool CanSeek => false;

    public override bool CanWrite => true;

    public override long Length => throw new NotSupportedException();

    public override long Position
    {
      get => throw new NotSupportedException();
      set => throw new NotSupportedException();
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
      throw new NotSupportedException();
    }

    public override void SetLength(long value)
    {
      throw new NotSupportedException();
    }

    #region -- Read --

    public override int Read(byte[] buffer, int offset, int count)
    {
      // ValueTask uses .GetAwaiter().GetResult() if necessary
      // https://github.com/dotnet/corefx/blob/f9da3b4af08214764a51b2331f3595ffaf162abe/src/System.Threading.Tasks.Extensions/src/System/Threading/Tasks/ValueTask.cs#L156
      return ReadAsync(new ArraySegment<byte>(buffer, offset, count)).Result;
    }

    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
    {
      var task = ReadAsync(new ArraySegment<byte>(buffer, offset, count));

      if (task.IsCompletedSuccessfully)
      {
        if (_cachedTask.Result != task.Result)
        {
          // Needs .AsTask to match Stream's Async method return types
          _cachedTask = task.AsTask();
        }
      }
      else
      {
        // Needs .AsTask to match Stream's Async method return types
        _cachedTask = task.AsTask();
      }

      return _cachedTask;
    }

    public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
    {
      var task = ReadAsync(buffer, offset, count, default(CancellationToken), state);
      return TaskToApm.Begin(task, callback, state);
    }

    public override int EndRead(IAsyncResult asyncResult)
    {
      return TaskToApm.End<int>(asyncResult);
    }

    private Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken, object state)
    {
      var tcs = new TaskCompletionSource<int>(state);
      var task = ReadAsync(buffer, offset, count, cancellationToken);
      task.ContinueWith((task2, state2) =>
      {
        var tcs2 = (TaskCompletionSource<int>)state2;
        if (task2.IsCanceled)
        {
          tcs2.SetCanceled();
        }
        else if (task2.IsFaulted)
        {
          tcs2.SetException(task2.Exception);
        }
        else
        {
          tcs2.SetResult(task2.Result);
        }
      }, tcs, cancellationToken);
      return tcs.Task;
    }

    private ValueTask<int> ReadAsync(ArraySegment<byte> buffer)
    {
      return _pipe.Reader.ReadAsync(buffer);
    }

    #endregion

    #region -- Write --

    public override void Write(byte[] buffer, int offset, int count)
    {
      //InternalWrite(_output, buffer, offset, count);
      _output.Write(new ReadOnlySpan<byte>(buffer, offset, count));
    }

    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
    {
      //return _pipe.Writer.WriteAsync(new ArraySegment<byte>(buffer, offset, count));
      if (cancellationToken.IsCancellationRequested)
      {
        return AsyncUtils.FromCanceled(cancellationToken);
      }

      _output.Write(new ReadOnlySpan<byte>(buffer, offset, count));
      //InternalWrite(_output, buffer, offset, count);
      return TaskConstants.Completed;
    }

    public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
    {
      //var task = WriteAsync(buffer, offset, count, default(CancellationToken), state);
      //return TaskToApm.Begin(task, callback, state);
      Write(buffer, offset, count);
      return new CompletedAsyncResult(callback, state);
    }

    public override void EndWrite(IAsyncResult asyncResult)
    {
      //TaskToApm.End(asyncResult);
      CompletedAsyncResult.End(asyncResult);
    }

    //private Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken, object state)
    //{
    //  var tcs = new TaskCompletionSource<object>(state);
    //  var task = WriteAsync(buffer, offset, count, cancellationToken);
    //  task.ContinueWith((task2, state2) =>
    //  {
    //    var tcs2 = (TaskCompletionSource<object>)state2;
    //    if (task2.IsCanceled)
    //    {
    //      tcs2.SetCanceled();
    //    }
    //    else if (task2.IsFaulted)
    //    {
    //      tcs2.SetException(task2.Exception);
    //    }
    //    else
    //    {
    //      tcs2.SetResult(null);
    //    }
    //  }, tcs, cancellationToken);
    //  return tcs.Task;
    //}

    //private static void InternalWrite(in WritableBuffer output, byte[] buffer, int offset, int count)
    //{
    //  var writer = new WritableBufferWriter(output);
    //  writer.Write(buffer, offset, count);
    //}

    #endregion

    public override void Flush()
    {
      _pipe.Writer.FlushAsync().GetResult();
    }

    public override Task FlushAsync(CancellationToken cancellationToken)
    {
      return _pipe.Writer.FlushAsyncAwaited(cancellationToken);
    }

    public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
    {
      return _pipe.Reader.CopyToAsync(destination, cancellationToken);
    }

#if NETCOREAPP
    public override void CopyTo(Stream destination, int bufferSize)
    {
      CopyToAsync(destination, bufferSize, default).GetAwaiter().GetResult();
    }
#endif

    protected override void Dispose(bool disposing)
    {
    }
    public override void Close()
    {
    }
  }
}
#endif